#!/usr/bin/perl -w
use strict;
use YAML;
use Data::Dumper;
use Getopt::Long;
use Config;
use threads;
use threads::shared;
use Thread::Queue;
use Time::Local;
use Date::Manip;
use Log::Log4perl;
use Schedule::Cron;
use Proc::Daemon;
use POSIX qw(setsid strftime getpid);
use FindBin ();
use File::Basename ();
use File::Spec::Functions;
use Time::HiRes qw(usleep);
use Archive::Zip qw( :ERROR_CODES :CONSTANTS );
use SecurityCenter;
use CenterFactory;
use DataCollectionChannel;
use CollectionChannelFactory;
use DataDumpingManager;
use DataAccessLayerFactory;
use StockInfoSet;

# configuration and log files
my $configuration_dir = "configuration/";
my $channel_config_file = $configuration_dir . "collection_channels.conf";
my $center_config_file = $configuration_dir . "security_centers.conf";
my $log_config_file = $configuration_dir . "log.conf";
my $database_config_file = $configuration_dir . "data_access.conf";
my $dump_file = ".mobisec.dat";
my $olog = "output.log";
my $elog = "error.log";

# process options
my %option;
if(!GetOptions('updaters=i' => \$option{'u'}, 
					'daemon' => \$option{'d'},
					'checkstatus' => \$option{'c'},
					'restart' => \$option{'r'},
					'stop' => \$option{'s'},
					'help'=> \$option{'h'})){
	usage_quit(1);
}
usage_quit(0) if $option{'h'} ;

# init the data dumping manager
my $data_dumping_manager = DataDumpingManager::new();

# stop or restart the service
if($option{'s'} || $option{'r'}){
	my $pid = $data_dumping_manager->load_dumped_data($dump_file, 'PID');
	if(defined($pid)){
		if($option{'s'}){
			system("kill -9 $pid 1>/dev/null 2>/dev/null");
			if ($? != 0) {
                print "Can't stop the service with error code: $? \n";               
            };
            exit;
		}
		if($option{'r'}){
			system("kill -HUP $pid 1>/dev/null 2>/dev/null");
			if ($? != 0) {
                print "Can't stop the service with error code: $? \n";               
            };
            exit;
		}
	}else{
		print "Can't identify the process id, exit\n";
		exit;
	}		
}

# init the logger
Log::Log4perl->init($log_config_file);
our $logger = Log::Log4perl::get_logger();
#touch log file
eval{
		my $appenders = Log::Log4perl->appenders();
		foreach my $appender_name (keys %{$appenders}){
			my $in;
			open($in , ">>", $appenders->{$appender_name}->filename());
			close($in);
		}
};
if($@){
	print "Can't access the log file $@";
	exit;
}
$logger->info('INFO|System started');

# check thread support
if($Config{useithreads}){
	# threads are supported
	$logger->warn('WARN|Threads are supported');	
}else{
	# threads aren't supported
	$logger->fatal('FALTAL|Threads are not supported. Exitting...');
	exit;
}
# catch sig-hup 
my $script = File::Basename::basename($0);
my $SELF = catfile $FindBin::Bin, $script;
# POSIX unmasks the sigprocmask properly
my $sigset = POSIX::SigSet->new();
my $action = POSIX::SigAction->new('sigHUP_handler',$sigset,&POSIX::SA_NODEFER);
POSIX::sigaction(&POSIX::SIGHUP, $action);

if($option{'d'}){ 	# daemonize the process
	# flush the buffer
	$| = 1;	
	# chdir '/home/nhchung'                 or die "Can't chdir to /: $!";
	open STDIN, '/dev/null'   or die "Can't read /dev/null: $!";
	open STDOUT, '>>'.$olog or die "Can't write to $olog: $!";
	open STDERR, '>>'.$elog or die "Can't write to $elog: $!";
	defined(my $pid = fork)   or die "Can't fork: $!";
	exit if $pid;
	setsid                    or die "Can't start a new session: $!";
	umask 0;
}

# init the database access manager
my $dal_manager = DataAccessLayerFactory::new();

# create collection channel factory
my $channel_manager =  CollectionChannelFactory::new();

# create center factory 
my $center_manager = CenterFactory::new();
# create queue and thread list
my %queue_list;
my %thread_list; # ($channel_name => $thr,...)

# create the scheduler
my $scheduler = new Schedule::Cron(\&dispatcher);

# dump process id
$data_dumping_manager->dump_data($dump_file, "PID", getpid());


# load channel configuration from a yaml file
my ($channels, $message) = $channel_manager->load_channels_from_yaml_file($channel_config_file);
if(!ref($channels)){
	$logger->error('ERROR|' . $message);
	exit;
}

# load center configuration from a yaml file
(my $centers, $message) = $center_manager->load_centers_from_yaml_file($center_config_file);
if(!ref($centers)){
	$logger->error('ERROR|' . $message);
	exit;
}

# add a cron jobs 
foreach my $center (@{$centers}){
	# threads are scheduled to run before openning time 3 minutes
	if($center->{START_TIME}->{MIN} < 3){
		my $min = 60 + $center->{START_TIME}->{MIN} - 3;
		my $hour = ($center->{START_TIME}->{HOUR} < 1)? 23 : ($center->{START_TIME}->{HOUR}-1); 
		$scheduler->add_entry($min . " " . $hour . " * * MON-FRI", \&create_threads, $center);
	}else{
		my $min = $center->{START_TIME}->{MIN} - 3;
		my $hour = $center->{START_TIME}->{HOUR};
		$scheduler->add_entry($min . " " . $hour . " * * MON-FRI", \&create_threads, $center);		
	}
	$center->{START_TIME}->{MIN} = '0' . $center->{START_TIME}->{MIN}  if($center->{START_TIME}->{MIN} =~ /^\d$/);	
	# check the current time and start threads if needed
	my $last = $center->{LAST};
	my $center_start_time = ParseDate($center->{START_TIME}->{HOUR} . ':' . $center->{START_TIME}->{MIN} . ':00' );	
	# calculate stop_time
	my $center_stop_time = DateCalc($center_start_time, "+ $last minutes");
	$scheduler->add_entry(UnixDate($center_stop_time, "%M"). " " . UnixDate($center_stop_time, "%H") . " * * MON-FRI", \&cleanup, $center);
	if(Date_Cmp(ParseDate("today"), DateCalc($center_start_time,"- 3 minutes")) >= 0 && Date_Cmp(ParseDate("today"),$center_stop_time) <= 0 ){
		create_threads($center);
	}
	
}

# run the cron	
$scheduler->run(nofork => 1);	

sub cleanup{
	my $center = shift;
	my $channels_by_center = $channel_manager->get_channels_by_center($center, $channels);
	foreach my $channel (@{$channels_by_center}){
		$thread_list{"COL_" . $center->code() . "_" . $channel->name()}->join();
		delete $thread_list{"COL_" . $center->code() . "_" . $channel->name()};
		for(my $i = 0; $i < $option{'u'}; $i++){ 
			if(exists $thread_list{"UPD_" . $center->code() . "_" . $channel->name() . "_$i"}){
				$thread_list{"UPD_" . $center->code() . "_" . $channel->name() . "_$i"}->join();
				delete $thread_list{"UPD_" . $center->code() . "_" . $channel->name() . "_$i"};			
			}
			delete $queue_list{"QUE_"  . $center->code() . "_" . $channel->name() . "_$i"};	
		}						
	}
	#time to refresh to release memory
	refresh();
}

#######################
## create threads	 ##
#######################
sub create_threads{
	# create threads that collect data from this $center
	my $center = shift;
	my $channels_by_center = $channel_manager->get_channels_by_center($center, $channels);
	foreach my $channel (@{$channels_by_center}){		
		for(my $i = 0; $i < $option{'u'}; $i++){
			my $stockinfo_queue = Thread::Queue->new();
			$queue_list{"QUE_" .  $center->code() . "_" . $channel->name() . "_$i"} = $stockinfo_queue; 
			if(! exists $thread_list{"UPD_"  . $center->code() . "_" . $channel->name() . "_$i"}){
				my $stock_thr = threads->create(\&update_stock_info, $channel->name(), $center->code(), $i);
				$thread_list{"UPD_"  . $center->code() . "_" . $channel->name() . "_$i"} = $stock_thr;			
			}	
		}
		
		if($center->{CODE} eq $channel->{SECURITY_CENTER_CODE}){
			my $thr = threads->create(\&collect_data, $channel, $center);
			$thread_list{"COL_" . $center->code() . "_"  . $channel->name()} = $thr;			
		}
	}	
	
}

sub collect_data{
	my $channel = shift;
	my $center = shift; 
	$logger->info('INFO| Spawn collecting channel ' . $channel->{NAME});
	#calculate start_time
	my $last = $center->{LAST};
	my $center_start_time = ParseDate($center->{START_TIME}->{HOUR} . ':' . $center->{START_TIME}->{MIN} . ':00' );
	# calculate stop_time
	my $center_stop_time = DateCalc($center_start_time, "+ $last minutes");
	if($center->code() eq 'HoSE'){
		if($option{'c'}){	# --checkstatus switch is on, check market status before running		
			$logger->info('INFO|' . $channel->name() . "| Market status checking...");
			while(1){
				# if the current time is later than the market closing time, exit;
				if(Date_Cmp(ParseDate("today"), $center_stop_time) > 0 ){
					$logger->info('INFO|COL_' . $center->code() . "_"  . $channel->name() . '| It\'s the market closing time .Exitting...');
					return;
				}
				my $market_status = $channel->get_market_status($center_start_time, $center_stop_time);
				$logger->info('INFO|' . $channel->name() . '| '. $market_status->{CODE} . '|' . $market_status->{MSG});
				if($market_status->{CODE} ne  'OPENING'){				
					sleep(5);
				}else{
					last;
				}
			};
		}
		# load and dump delay time
		my $delay_time = $data_dumping_manager->load_dumped_data($dump_file, $channel->name() . "_DELAY_TIME");
		if(!$delay_time || $delay_time eq '' ){
			$channel->{DELAY_TIME} = UnixDate(ParseDate("today"), "%s") - UnixDate($center_start_time, "%s");
			$data_dumping_manager->dump_data($dump_file, $channel->name() . "_DELAY_TIME", $channel->{DELAY_TIME});				  
		}else{
			$channel->{DELAY_TIME} = $data_dumping_manager->load_dumped_data($dump_file, $channel->name() . "_DELAY_TIME");
		}
		
		# market now open
		$logger->info('INFO|' . $channel->name() . "| Market was openned, start to collect data");
		my $cached_sys_sequence = 0;
		my $queue_idx = 0;
		while(1){			
			my $sys_seq;
			if($cached_sys_sequence == 0){ # thread start, $cached_sys_sequence = 0
				 $sys_seq = $data_dumping_manager->load_dumped_data($dump_file, $channel->name(). "_SEQ");
				 if(!$sys_seq){
					$sys_seq = $cached_sys_sequence;					
				 }					 
			}else{
				$sys_seq = $cached_sys_sequence;				
			}		
			$data_dumping_manager->dump_data($dump_file, $channel->name() . "_SEQ", ++$sys_seq);
			$cached_sys_sequence = $sys_seq; 
						
			(my $stock_info_set, my $status) = $channel->get_stock_info_set($sys_seq);
			# choose a queue to deliver data
			$queue_idx ++;
			if($queue_idx >= $option{'u'}){
				$queue_idx = 0;
			}
			my $sleep_time = $channel->{COLLECTING_DURATION} - $stock_info_set->{CHANNEL_INFO}->get('PROCESSING_TIME');
			sleep(($sleep_time)?$sleep_time:0);
	        if($status->{CODE} eq 'TIMEOUT' || $status eq 'FAILED'){
				$logger->error('ERROR|' . $channel->name() . "|" . $sys_seq . '| ' . $status->{MSG});				
				$queue_list{"QUE_"  . $center->code() . "_" . $channel->name() . "_" . $queue_idx}->enqueue($stock_info_set);
				next;								
			}
			# fetched, put to queue
			$logger->info('INFO|COL_' . $center->code() . "_"  . $channel->name() . "|" . $sys_seq . "| " . $status->{MSG});
			
			# compare current market time with stop time 
			if(Date_Cmp(ParseDate($stock_info_set->{MARKET_INFO}->get('Time')), $center_stop_time) > 0 ){
				# deliver undef to all queues
				for(my $upd_thr = 0; $upd_thr < $option{'u'}; $upd_thr++){
					$queue_list{"QUE_" . $center->code() . "_" . $channel->name() . "_" . $upd_thr}->enqueue(undef);
				} 
				last;
			}else{				
				$queue_list{"QUE_"  . $center->code() . "_" . $channel->name() . "_" . $queue_idx}->enqueue($stock_info_set);
				next;
			}										
		}
		$logger->info('INFO|COL_' . $center->code() . "_"  . $channel->name() . '| Market was closed.Exitting...');
		return;
	}
	
}

sub update_stock_info{
	my ($dal, $dal_msg) = $dal_manager->load_data_access_layer_from_yaml_file($database_config_file);
	if(!$dal){	 	
		$logger->fatal('FATAL|Can\'t init the database access object, check the database configuration file. Exitting...');
		exit;
	}	
	my $channel_name = shift;
	my $center_code = shift;
	my $thr_num = shift;
	while (1) {
		my $stock_info_set = $queue_list{"QUE_"  . $center_code . "_" . $channel_name . "_" . $thr_num}->dequeue();
		if(defined($stock_info_set)){
			bless($stock_info_set, "StockInfoSet");
			my ($status, $time) = $dal->insert_stock_info_set($stock_info_set);
			if($status->{CODE} ne 'OK'){
				$logger->error('ERROR|UPD_' . $center_code . "_" . $channel_name . "|" . $thr_num .  "|" . $status->{MSG} . '| ' . $stock_info_set->from_channel() . '| ' . $stock_info_set->sys_sequence() . "| Market time: " . $stock_info_set->market_time());
				# put the $stock_info_set to the queue again
				# $stockinfo_queue->enqueue($stock_info_set);			
			}else{
				$logger->info('INFO|UPD_' . $center_code . "_" . $channel_name . "|" . $thr_num .  "|" . $status->{MSG} . '| ' . $stock_info_set->from_channel() . '| ' . $stock_info_set->sys_sequence() . "| Market time: " . $stock_info_set->market_time());
			}
			$stock_info_set->delete();									
		}else{			
			last;
		}				   	
    }   
}

sub reset_dumped_datas{
	if(ref($data_dumping_manager)){
		$data_dumping_manager->reset_all_dumped_datas($dump_file);		
	}
} 

sub dispatcher {  
   
}

sub sigHUP_handler {
		$logger->warn('WARN| Got SIGHUP. System restarting...');
		my $args = "";
		if($option{'u'}){
			$args .= " -u " . $option{'u'}; 
		}
		if($option{'d'}){
			$args .= " -d "; 
		}
		if($option{'c'}){
			$args .= " -c ";
		}
		exec($SELF . $args) or die "Couldn't restart: $!\n";
		exit;
}

sub refresh{
	# archive the log files
	my ($status, $message) = archive_log_files();
	if($status ne 'OK'){
		$logger->fatal('FATAL|' . $message);
		exit;
	}
	# refresh the service
	sigHUP_handler();
}

sub archive_log_files{
	my $arch_file ="archive/" . strftime("%d%m%Y", localtime(time)) . ".zip";
	my $zip = Archive::Zip->new();
	eval{
		my $appenders = Log::Log4perl->appenders();
		foreach my $appender_name (keys %{$appenders}){
			$zip->addFile($appenders->{$appender_name}->filename());
		}
		$zip->addFile($elog);
		$zip->addFile($olog);
		$zip->addFile($dump_file);
			
		unless ( $zip->writeToFileNamed($arch_file) == AZ_OK ) {
		   die('Can\'t write to the archive file');	   
		};		
		unlink $elog,$olog, $dump_file;
		foreach my $appender_name (keys %{$appenders}){
			unlink $appenders->{$appender_name}->filename();
		}
	};
	if($@){
		return ('FAILED', $@);
	}else{
		return ('OK',"Archived successfully");
	}	
}

sub usage_quit {
  # Emit usage message, then exit with given error code.
  print <<"END_OF_MESSAGE"; exit($_[0] || 0);
Usage:
$0  [switches]  
     
Switches:
 -h or --help			display this help message
 -d or --daemon		run as a daemon
 -c or --checkstatus		check market status before running
 -u or --updaters			specify the number of database updating threads  
 -s or --stop			stop the process
 -r or --restart			restart the process
END_OF_MESSAGE
}
